package com.example.test15middleware.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @Author: fxy
 * @Date: 2022/03/17/17:25
 * @Description:
 */
@Component
public class KafkaConsumer {
    @KafkaListener(topics = "yun_test",groupId = "1")
    public void consumer(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
        Optional message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            Object msg = message.get();
            System.out.println("group_test 消费了： Topic:" + topic + ",Message:" + msg);
        }
    }
}
